package com.atguigu.orderpay_detect

import java.util

import org.apache.flink.cep.{PatternSelectFunction, PatternTimeoutFunction}
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

/**
  * 1
  * @author: yangShen
  *         cep底层实现是使用了一个状态机
  * @Description: 订单支付超时分析 cep
  *              优点：可以来一条处理一条
  * @Date: 2020/5/9 14:05 
  */
//定义输入订单事件的样例类
case class OrderEvent( orderId: Long, eventType: String, txId: String, eventTime: Long)
//定义输出结果的样例类
case class OrderResult( orderId: Long, resultMsg: String)

object OrderTimeout {
  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    environment.setParallelism(1)

    //1.读取订单数据
    val resource = getClass.getResource("/OrderLog.csv")
//    val orderEventStream = environment.readTextFile(resource.getPath) //文件读取
    val orderEventStream = environment.socketTextStream("10.16.26.16", 7777)  //流式读取：nc -lk 7777
      .map(data => {
        val dataArray = data.split(",")
        OrderEvent(dataArray(0).trim.toLong, dataArray(1).trim, dataArray(2).trim, dataArray(3).trim.toLong)
      })
      //设置时间戳
      //方式一(简单实现)：升序数据，不设置延时 到点就发车，从字段中抽取
      .assignAscendingTimestamps(_.eventTime * 1000L)
      .keyBy(_.orderId)

    //2. 定义一个匹配模式
    val orderPayPattern = Pattern.begin[OrderEvent]("begin").where(_.eventType == "create")
      //宽松近邻
      .followedBy("pay").where(_.eventType == "pay")
      .within(Time.minutes(15))   //15分钟

    //3.把模式应用到stream上，得到一个pattern stream
    val patternStream = CEP.pattern(orderEventStream, orderPayPattern)

    //4.调用select方法，提取事件序列，超时的事件做报警提示
    //定义侧输出流标签
    val orderTimeoutOutputTag: OutputTag[OrderResult] = new OutputTag[OrderResult]("orderTimeout")

    //                            参数(侧输出流tag, 超时事件, 正常匹配的订单支付事件)
    val resultStream = patternStream.select(orderTimeoutOutputTag, new OrderTimeoutSelect(), new OrderPaySelect())

    resultStream.print("payed")
    resultStream.getSideOutput(orderTimeoutOutputTag).print("timeout")

    environment.execute("order timeout job")

  }
}
//自定义超时事件序列处理函数
class OrderTimeoutSelect() extends PatternTimeoutFunction[OrderEvent, OrderResult]{
  override def timeout(map: util.Map[String, util.List[OrderEvent]], timeoutTimestamp: Long): OrderResult = {
    //超时的数据只能在begin模式中，不能在pay模式中
    val timeoutOrderId = map.get("begin").iterator().next().orderId
    OrderResult(timeoutOrderId, "timeout")
  }
}

//自定义正常支付事件序列处理函数
class OrderPaySelect() extends PatternSelectFunction[OrderEvent, OrderResult]{
  override def select(map: util.Map[String, util.List[OrderEvent]]): OrderResult = {
    val payedOrderId = map.get("pay").iterator().next().orderId
    OrderResult(payedOrderId, "successfully")
  }
}
